package org.timewindow

import org.FlinkStreamApp
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.diysource.SensorSource

/**
 * description ：计算每个传感器15s窗口中的温度最小值
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/23 10:06
 * modified By ：
 * version:    : 1.0
 */
object ReduceFunctionExample extends FlinkStreamApp {
  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    // 获取源数据
    val source = environment.addSource(new SensorSource)

    // 计算
    val minTempStream = source
      .map(l => (l.id, l.temperature))
      .keyBy(_._1)
      .timeWindow(Time.seconds(15))
      .reduce((x, y) => (x._1, x._2.min(y._2)))

    // 输出结果
    minTempStream.print
  }
}
